package org.jsets.rpc.protocol.hessian;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.SocketAddress;
import java.util.Base64;
import java.util.Base64.Decoder;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;

import org.jsets.rpc.conf.SerivceConfig;
import org.jsets.rpc.processor.SerivceExporter;
import org.jsets.rpc.processor.ServiceMetaData;
import org.jsets.rpc.processor.ServiceProcessor;
import org.jsets.rpc.transfer.RequestWrapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.caucho.hessian.io.AbstractHessianInput;
import com.caucho.hessian.io.AbstractHessianOutput;
import com.caucho.hessian.io.Hessian2Input;
import com.caucho.hessian.io.HessianFactory;
import com.caucho.hessian.io.HessianInputFactory;
import com.caucho.hessian.io.SerializerFactory;
import com.caucho.services.server.ServiceContext;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.DefaultHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.util.CharsetUtil;

/**
 * @ClassName: HessionHandler
 * @Description: HESSION 编码器
 * @author wangjie
 * @date 2015年7月11日 下午3:30:52
 */
@Sharable
public class HessionHandler extends ChannelInboundHandlerAdapter  {
	private final static Logger logger = LoggerFactory.getLogger(HessionHandler.class.getSimpleName());
	
	@SuppressWarnings("unused")
	private SerivceConfig serivceConfig;
	private ExecutorService threadpool;
	
	@SuppressWarnings("unused")
	private boolean readingChunks;
	
	private FullHttpRequest request;
	private HessianInputFactory inputFactory = new HessianInputFactory();
	private HessianFactory hessianFactory = new HessianFactory();
	private SerializerFactory serializerFactory;
	
	private final StringBuilder buf = new StringBuilder();
	private final Decoder decoder = Base64.getDecoder();

	/**
	 * 构造函数
	 * @param serivceConfig
	 * @param threadpool
	 */
	public HessionHandler(SerivceConfig serivceConfig, ExecutorService threadpool) {
		this.threadpool = threadpool;
		this.serivceConfig = serivceConfig;
	}
	
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception{
		logger.info(":::::: connect: {}", ctx.channel().remoteAddress());
	}
	
	@Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		logger.info(":::::: disconect: {}", ctx.channel().remoteAddress());
    }
	
	@Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		logger.info("recv: {}\r\n==================================", msg);
		
		this.request = (FullHttpRequest) msg;
		String uri = request.uri();
		if (uri.endsWith("/")) 
			uri = uri.substring(0, uri.length() - 1);
		
		String serviceName = uri.substring(uri.lastIndexOf("/") + 1);
		SocketAddress remoteAddress = ctx.channel().remoteAddress();
		String ipAddress = remoteAddress.toString().split(":")[0];
		request.headers().set("Client-IP", ipAddress.substring(1));

		FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
		ByteArrayOutputStream os = new ByteArrayOutputStream();
		
		handleService(serviceName, request, response, os, ctx);
	}

	/**
	 * 处理请求
	 * @param serviceName
	 * @param request
	 * @param response
	 * @param os
	 * @param _ctx
	 * @throws Exception
	 */
	private void handleService(final String serviceName, 
			                   final FullHttpRequest request, 
			                   final FullHttpResponse response, 
			                   final ByteArrayOutputStream os, 
			                   final ChannelHandlerContext _ctx
			                   )throws Exception {
		try {
			threadpool.execute(new Runnable() {
				public void run() {
					try {
						service(serviceName, request, response, os);
					} catch (Exception e1) {
						logger.error(e1.getMessage(), e1);
					}
					
					if (HttpUtil.is100ContinueExpected(request)) 
						send100Continue(_ctx);
					
					writeResponse(_ctx, response, os);
				}
			});
		} catch (RejectedExecutionException exception) {
			logger.error("server threadpool full, threadpool maxsize is: {}", ((ThreadPoolExecutor) threadpool).getMaximumPoolSize());
		}
	}
	
	/**
	 * 
	 * @param serviceName
	 * @param req
	 * @param res
	 * @param os
	 */
	public void service(String serviceName, 
			            FullHttpRequest req, 
			            HttpResponse res, 
			            ByteArrayOutputStream os
			            ) {
		byte[] bytes = new byte[req.content().readableBytes()]; // .toString().getBytes();
																// request
																// contentd
		req.content().readBytes(bytes);
		InputStream is = new ByteArrayInputStream(bytes);
		serializerFactory = getSerializerFactory();
		String[] authLink = getUsernameAndPassword(req);
		
		String username = authLink[0].equals("") ? null : authLink[0];
		String password = authLink[1].equals("") ? null : authLink[1];
		String clientIP = req.headers().get("Client-IP");
		
		RequestWrapper rw = new RequestWrapper(username, password, clientIP, serviceName);
		invoke(rw, is, os, serializerFactory);
	}
	
	/**
	 * 用于客户端在发送POST数据给服务器前，征询服务器情况，看服务器是否处理POST的数据，
	 * 如果不处理，客户端则不上传POST数据，
	 * 如果处理，则POST上传数据。
	 * 在现实应用中，通过在POST大数据时，才会使用100-continue协议。
	 * @param ctx
	 */
	private void send100Continue(ChannelHandlerContext ctx) {
		DefaultHttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE);
		if (request.content().hasArray()) 
			buf.append(request.content().toString(CharsetUtil.UTF_8));
		
		ChannelFuture future = ctx.channel().write(response);
		boolean keepAlive = HttpUtil.isKeepAlive(request);
		if (!keepAlive) 
			future.addListener(ChannelFutureListener.CLOSE);
	}

	/**
	 * 回复数据
	 * @param ctx
	 * @param response
	 * @param os
	 */
	private void writeResponse(ChannelHandlerContext ctx, 
			                   FullHttpResponse response, 
			                   ByteArrayOutputStream os
			                   ) {
		boolean keepAlive = HttpUtil.isKeepAlive(request);
		
		response.content().writeBytes(os.toByteArray());
		response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");
		response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
		
		logger.info("response: {}, {}", os.toString().length(), os.toString());

		ChannelFuture future = ctx.channel().writeAndFlush(response);
		
		if (!keepAlive) 
			future.addListener(ChannelFutureListener.CLOSE);
	}

	public void setSerializerFactory(SerializerFactory _factory) {
		serializerFactory = _factory;
	}

	public SerializerFactory getSerializerFactory() {
		if (serializerFactory == null)
			serializerFactory = new SerializerFactory();

		return serializerFactory;
	}

	private String[] getUsernameAndPassword(FullHttpRequest _req) {
		String auths = request.headers().get("Authorization");
		if (auths == null) {
			String str[] = { "", "" };
			return str;
		}
		String auth[] = auths.split(" ");
		String bauth = auth[1];
		String dauth = new String(decoder.decode(bauth));
		String authLink[] = dauth.split(":");
		return authLink;
	}

	/**
	 * 分解
	 * @param _wrapper
	 * @param _is
	 * @param _os
	 * @param _serializerFactory
	 */
	protected void invoke(RequestWrapper _wrapper, 
			              InputStream _is, 
			              OutputStream _os, 
			              SerializerFactory _serializerFactory
			              ) {
		AbstractHessianInput in = null;
		AbstractHessianOutput out = null;
		try {
			HessianInputFactory.HeaderType header = inputFactory.readHeader(_is);
			switch (header) {
				case CALL_1_REPLY_1:
					in = hessianFactory.createHessianInput(_is);
					out = hessianFactory.createHessianOutput(_os);
					break;
					
				case CALL_1_REPLY_2:
					in = hessianFactory.createHessianInput(_is);
					out = hessianFactory.createHessian2Output(_os);
					break;
					
				case HESSIAN_2:
					in = hessianFactory.createHessian2Input(_is);
					in.readCall();
					out = hessianFactory.createHessian2Output(_os);
					break;
					
				default:
					throw new IllegalStateException(header + " is an unknown Hessian call");
			}
			
			if (_serializerFactory != null) {
				in.setSerializerFactory(_serializerFactory);
				out.setSerializerFactory(_serializerFactory);
			}
			
			invoke(_wrapper, in, out);
		} catch (Exception e) {
			try {
				out.writeFault("ServiceException", e.getMessage(), e);
			} catch (IOException e1) {
				e1.printStackTrace();
			}
		} finally {
			try {
				in.close();
				out.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	public void invoke(RequestWrapper _wrapper, AbstractHessianInput in, AbstractHessianOutput out) throws Exception {
		ServiceContext context = ServiceContext.getContext();
		String serviceName = _wrapper.getServiceName();
		in.skipOptionalCall();
		String header;
		while ((header = in.readHeader()) != null) {
			Object value = in.readObject();
			context.addHeader(header, value);
		}
		
		ServiceMetaData metaData = SerivceExporter.getServiceMetaData(serviceName);
		if (metaData == null) {
			logger.error("service {} can't find.", serviceName);
			out.writeFault("NoSuchService", "service " + serviceName + " can't find.", null);
			out.close();
			return;
		}
		
		String methodName = in.readMethod();
		int argLength = in.readMethodArgLength();
		Method method = metaData.getMethod(methodName + "__" + argLength);
		if (method == null) 
			method = metaData.getMethod(methodName);
		
		if (method == null) {
			out.writeFault("NoSuchMethod", "service[" + methodName + "]'s method " + methodName + " cannot find", null);
			out.close();
			return;
		}
		
		Class<?>[] argTypes = method.getParameterTypes();
		Object[] argObjs = new Object[argTypes.length];
		for (int i = 0; i < argTypes.length; i++) 
			argObjs[i] = in.readObject(argTypes[i]);
		
		_wrapper.setMethodName(method.getName());
		_wrapper.setArgs(argObjs);
		_wrapper.setArgsTypes(argTypes);
		
		if (argLength != argObjs.length && argLength >= 0) {
			out.writeFault("NoSuchMethod", "service[" + methodName + "]'s method " + methodName
					+ " argument length mismatch, received length=" + argLength, null);
			out.close();
			return;
		}
		
		Object result = null;
		try {
			result = ServiceProcessor.handleRequest(_wrapper);
		} catch (Exception e) {
			Throwable e1 = e;
			if (e1 instanceof InvocationTargetException)
				e1 = ((InvocationTargetException) e).getTargetException();

			logger.debug(this + " " + e1.toString(), e1);
			result = e;
			out.writeFault("ServiceException", e1.getMessage(), e1);
			out.close();
			return;
		}
		
		in.completeCall();
		out.writeReply(result);
		out.close();
	}

	protected Hessian2Input createHessian2Input(InputStream is) {
		return new Hessian2Input(is);
	}

}
